package com.gzx.bi.core.kafka;

import com.gzx.bi.modular.agent.service.GzxAgentEffectService;
import com.gzx.bi.modular.team.service.GzxTeamEffectService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import vip.xiaonuo.common.kafka.KafkaTopicConstant;
import vip.xiaonuo.common.tenant.TenantContext;

import java.util.List;

@Slf4j
@Component
@RequiredArgsConstructor
public class DataBoardKafkaConsumer {

    //数据看板——团队效率
    private final GzxTeamEffectService gzxTeamEffectService;

    //数据看板——坐席效率
    private final GzxAgentEffectService gzxAgentEffectService;

    /**
     * 项目，用户部分数据每小时同步一次，数据看板相关业务直接关联查询即可
     */
    @KafkaListener(topics = KafkaTopicConstant.DATA_BOARD_TOPIC,
            groupId = "my-consumer-group", containerFactory = "batchFactory")
    public void saveBatchDashboard(List<ConsumerRecord<String, String>> recordList, Acknowledgment ack) {
        log.info("从MQ中批量获取话单 data-dashboard：本批次处理event数量{}", recordList.size());
        log.info("Kafka收到消息 data-dashboard：{}", recordList);
        try {
            TenantContext.disableTenantFilter();
            //坐席效率报表
            gzxAgentEffectService.saveAgentBatch(recordList);
            //团队效率报表
            gzxTeamEffectService.saveTeamBatch(recordList);
            // 业务处理完毕再手动提交offset
            ack.acknowledge();
        } catch (Exception e) {
            log.error("当前批次消息处理失败 data-dashboard：", e);
            throw e;
        } finally {
            log.info("Kafka处理完毕 data-dashboard：{}", recordList);
            TenantContext.enableTenantFilter();
            TenantContext.clear();
        }
    }

}